package pulsar_topic_mover

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"
	"go_19/core/logrus"
	"go_19/global"
	"go_19/initialize"
	"time"
)

type TopicMoverStruct struct {
	FromTopic initialize.TopicConsumer `json:"from_topic"`
	ToTopic   initialize.TopicProducer `json:"to_topic"`
}

var ToTopicProducer pulsar.Producer

var ToTopicProducerLog struct {
	FromMsgId      string `json:"from_msg_id" mapstructure:"from_msg_id"`           //来源msgId
	FromTopic      string `json:"from_topic"`                                       //来源topic
	ToTopic        string `json:"to_topic"`                                         //发布topic
	PublishedMsgId string `json:"published_msg_id" mapstructure:"published_msg_id"` //发布msgId
	PublishedMsg   string `json:"published_msg" mapstructure:"published_msg"`       //发布消息内容
}

func TopicMover(topicMover TopicMoverStruct) {
	client := initialize.PulsarClient(topicMover.FromTopic.ClientOption)
	consumerMessage := make(chan pulsar.ConsumerMessage, 1)

	isUnsubscribe := 0
getConsumer:
	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:                       topicMover.FromTopic.Topic,
		SubscriptionName:            topicMover.FromTopic.SubscriptionName,
		SubscriptionInitialPosition: pulsar.SubscriptionInitialPosition(topicMover.FromTopic.SubscriptionInitialPosition),
		Type:                        pulsar.SubscriptionType(topicMover.FromTopic.SubscriptionType),
		Name:                        topicMover.FromTopic.ConsumerName,
		MessageChannel:              consumerMessage,
	})
	if err != nil {
		global.GVA_LOG.Error("TopicMover consumer err", err)
		panic("TopicMover consumer err")
	}

	//earliest 移除订阅者 重新订阅
	if topicMover.FromTopic.SubscriptionInitialPosition == 1 && isUnsubscribe == 0 {
		consumer.Unsubscribe()
		isUnsubscribe = 1
		goto getConsumer
	}

	toTopicClient := initialize.PulsarClient(topicMover.ToTopic.ClientOption)
	ToTopicProducer, err = toTopicClient.CreateProducer(pulsar.ProducerOptions{
		Topic: topicMover.ToTopic.Topic,
		Name:  topicMover.ToTopic.ProducerName,
	})
	if err != nil {
		global.GVA_LOG.Error("TopicMover producer err1", err)
		panic("TopicMover consumer err")
	}
	PublishedLogName := "published_" + topicMover.ToTopic.ClientOption.LogName
	publishedLogger := logrus.CreateSimpleLogger(PublishedLogName)
	for cm := range consumerMessage {
		if msgId, err := ToTopicProducer.Send(context.Background(), &pulsar.ProducerMessage{
			Payload:      cm.Payload(),
			DeliverAfter: time.Second * time.Duration(topicMover.ToTopic.DeliverAfter),
		}); err != nil {
			global.GVA_LOG.Error("TopicMover producer publish err", err)
			panic("TopicMover producer publish err")
		} else {
			fromMsgId := cm.ID()
			global.GVA_LOG.Info("From message", fromMsgId, "Published message: ", msgId)
			logStruct := ToTopicProducerLog
			logStruct.FromTopic = topicMover.FromTopic.Topic
			logStruct.ToTopic = topicMover.ToTopic.Topic

			logStruct.FromMsgId = fmt.Sprintf("%v", fromMsgId)
			logStruct.PublishedMsgId = fmt.Sprintf("%v", msgId)
			logStruct.PublishedMsg = string(cm.Payload())
			publishedLogJson, err := json.Marshal(logStruct)
			if err == nil {
				publishedLogger.Info(string(publishedLogJson))
			}
			consumer.Ack(cm.Message)
		}

	}
	defer func() {
		consumer.Close()
		ToTopicProducer.Close()
	}()
}
